package org.example.ItemCF.src.recommend;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.example.ItemCF.src.hdfs.HdfsDAO;

/**
 * 第三步：用户向量拆分。该步骤将用户-物品偏好向量拆分成物品-用户偏好对。
 * 输入格式为用户ID和其对应偏好向量，输出为物品ID和对应用户偏好。
 */
public class Step3Driver {

    /**
     * 执行MapReduce任务，将用户向量拆分成物品-用户偏好对。
     * @param path 包含输入输出路径的映射表。
     * @throws IOException 当发生I/O错误时抛出。
     * @throws ClassNotFoundException 当尝试加载曾在作业中使用过的自定义类时抛出。
     * @throws InterruptedException 当作业被中断时抛出。
     */
    public static void run1(Map<String, String> path) throws IOException, ClassNotFoundException, InterruptedException {

        Configuration conf = new Configuration();

        // 获取输入输出路径
        String input = path.get("Step3Input");
        String output = path.get("Step3Output");

        // 初始化HDFS操作
        HdfsDAO hdfs = new HdfsDAO(Recommend.HDFS, conf);
        hdfs.rmr(output); // 删除输出目录下的旧文件

        // 设置作业配置
        Job job = Job.getInstance(conf, "RecommendStep31");
        job.setJarByClass(Step3Driver.class); // 指定作业的JAR包
        job.setMapOutputKeyClass(IntWritable.class); // 设置Map输出键的类型
        job.setMapOutputValueClass(Text.class); // 设置Map输出值的类型

        job.setOutputKeyClass(IntWritable.class); // 设置Reduce输出键的类型
        job.setOutputValueClass(Text.class); // 设置Reduce输出值的类型

        job.setMapperClass(Step3UserVectorSplitterMapper.class); // 指定Map函数类
        // job.setReducerClass(Step1ToUserVectorReducer.class); // 注释了Reducer类的设置，当前任务不使用Reducer
        job.setNumReduceTasks(0); // 设置Reducer任务数为0
        FileInputFormat.addInputPath(job, new Path(input)); // 设置输入路径
        FileOutputFormat.setOutputPath(job, new Path(output)); // 设置输出路径
        // 等待作业完成
        if (!job.waitForCompletion(true))
            return;
    }

}
